package com.atguigu.edu.realtime220815.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime220815.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime220815.bean.PlayVideoTimesBean;
import com.atguigu.edu.realtime220815.util.ClickHouseUtils;
import com.atguigu.edu.realtime220815.util.DateFormatUtil;
import com.atguigu.edu.realtime220815.util.KafkaUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

import static org.apache.flink.api.common.time.Time.seconds;

/**
 * @Author: rym
 * @Description: TODO
 * @DateTime: 2023/2/16 14:32
 **/
public class DwsPlayVideoTimesWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        /*env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000);
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30), seconds(3)));
        env.setStateBackend( new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop101:8020/edu/ck");
        System.setProperty("HADOOP_USER_NAME","atguigu");*/

        String topic = "dwd_traffic_appVideo_log";
        String groupId = "DwsPlayVideoWindow_groupId";
        FlinkKafkaConsumer<String> flinkKafkaConsumer = KafkaUtils.getFlinkKafkaConsumer(topic, groupId);
        DataStreamSource<String> kfkDS = env.addSource(flinkKafkaConsumer);


        SingleOutputStreamOperator<JSONObject> mapDS = kfkDS.map(JSON::parseObject);


        SingleOutputStreamOperator<PlayVideoTimesBean> beanDS = mapDS.map(
                new MapFunction<JSONObject, PlayVideoTimesBean>() {
                    @Override
                    public PlayVideoTimesBean map(JSONObject value) throws Exception {
                        String sid = value.getJSONObject("common").getString("sid");
                        Long ts = value.getLong("ts");
                        String video_id = value.getJSONObject("appVideo").getString("video_id");

                        PlayVideoTimesBean playVideoTimesBean = PlayVideoTimesBean.builder()
                                .sidSet(new HashSet<>(Collections.singleton(sid)))
                                .ts(ts)
                                .videoId(video_id)
                                .build();
                        return playVideoTimesBean;
                    }
                }
        );
        //beanDS.print();

        SingleOutputStreamOperator<PlayVideoTimesBean> withwaterDS = beanDS.assignTimestampsAndWatermarks(WatermarkStrategy.<PlayVideoTimesBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<PlayVideoTimesBean>() {
                    @Override
                    public long extractTimestamp(PlayVideoTimesBean element, long recordTimestamp) {
                        return element.getTs();
                    }
                }));

        KeyedStream<PlayVideoTimesBean, String> keyByDS = withwaterDS.keyBy(value -> value.getVideoId());

        WindowedStream<PlayVideoTimesBean, String, TimeWindow> windowDS = keyByDS.window(
                TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10))
        );

        SingleOutputStreamOperator<PlayVideoTimesBean> reduceDS = windowDS.reduce(
                new ReduceFunction<PlayVideoTimesBean>() {
                    @Override
                    public PlayVideoTimesBean reduce(PlayVideoTimesBean value1, PlayVideoTimesBean value2) throws Exception {
                        value1.getSidSet().addAll(value2.getSidSet());
                        return value1;
                    }
                },
                new WindowFunction<PlayVideoTimesBean, PlayVideoTimesBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<PlayVideoTimesBean> input, Collector<PlayVideoTimesBean> out) throws Exception {
                        long start = window.getStart();
                        long end = window.getEnd();
                        long now = System.currentTimeMillis();
                        for (PlayVideoTimesBean playVideoTimesBean : input) {
                            playVideoTimesBean.setStt(DateFormatUtil.toYmdHms(start));
                            playVideoTimesBean.setEdt(DateFormatUtil.toYmdHms(end));
                            playVideoTimesBean.setTs(now);
                            playVideoTimesBean.setCount((long) playVideoTimesBean.getSidSet().size());
                            out.collect(playVideoTimesBean);
                        }
                    }
                }


        );
        //reduceDS.print();

        SingleOutputStreamOperator<PlayVideoTimesBean> withDim_video_infoDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<PlayVideoTimesBean>("dim_video_info") {
                    @Override
                    public void join(JSONObject dimInfoJsonObj, PlayVideoTimesBean obj) {
                        String chapter_id = dimInfoJsonObj.getString("chapter_id".toUpperCase());
                        obj.setChapterId(chapter_id);
                    }

                    @Override
                    public String getKey(PlayVideoTimesBean obj) {
                        return obj.getVideoId();
                    }
                }, 60, TimeUnit.SECONDS
        );
        //withDim_video_infoDS.print();

        SingleOutputStreamOperator<PlayVideoTimesBean> withDim_chapter_infoDS = AsyncDataStream.unorderedWait(
                withDim_video_infoDS,
                new DimAsyncFunction<PlayVideoTimesBean>("dim_chapter_info") {
                    @Override
                    public void join(JSONObject dimInfoJsonObj, PlayVideoTimesBean obj) {
                        String chapter_name = dimInfoJsonObj.getString("chapter_name".toUpperCase());
                        obj.setChapterName(chapter_name);
                    }

                    @Override

                    public String getKey(PlayVideoTimesBean obj) {
                        return obj.getChapterId();
                    }
                }, 60, TimeUnit.SECONDS
        );
        withDim_chapter_infoDS.print(">>>");

        withDim_chapter_infoDS.addSink(ClickHouseUtils.getClickHouseSinkFunction(
                "insert into dws_play_video_times_window values(?,?,?,?,?,?,?)"
        ));

        env.execute();



    }
}
